/*
Copyright 2010 Dmitry Barashev

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
 */
package net.barashev.seclub.io;

import index.CalculatorOfWorkingWithDisk;

import java.io.IOException;
import java.nio.BufferOverflowException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.ReadableByteChannel;
import java.util.LinkedList;
import java.util.List;
import java.util.ListIterator;

import net.barashev.seclub.io.file.VFile;

/**
 * Implements buffer manager, whose responsibility is to cache disk blocks in memory
 * pull blocks into buffers as needed and compact buffers which aren't used anymore.
 * @author Dmitry Barashev
 */
public class BufferManager {
    private final ByteBuffer mainBuffer;
    private final int blockSize;
    private final List<Slot> slotList = new LinkedList<Slot>();
    private final int bufferCount;

    /**
     * Creates a buffer manager capable of holding up to {@code bufferCount} buffers
     * {@code blockSize} bytes each.
     * @param blockSize
     * @param bufferCount
     */
    public BufferManager(int blockSize, int bufferCount) {
        this.bufferCount = bufferCount;
        int mainBufferSize = bufferCount*blockSize;
        this.mainBuffer = ByteBuffer.allocate(mainBufferSize);
        this.blockSize = blockSize;

    }

    public <T, M extends DataMarshaller<T>> IteratingSlot<T, DataMarshaller<T>> allocateIteratingSlot(
            int bufferCount, ReadableByteChannel channel, M marshaller) {
    	CalculatorOfWorkingWithDisk.callAllocateIteratingSlot();
    	
        ByteBuffer slice = allocateSlice(bufferCount*this.blockSize);
        IteratingSlot<T, DataMarshaller<T>> result = new IteratingSlot<T, DataMarshaller<T>>(
                slice, channel, marshaller, this);
        this.slotList.add(result);
        return result;
    }

    /**
     * Allocates a new iterating slot of {@code bufferCount} buffers for iterating over {@code file}.
     * Compacts released buffers if necessary.
     * @throws BufferOverflowException if there are no {@code bufferCount} available slots.
     */
    public <T, M extends DataMarshaller<T>> IteratingSlot<T, DataMarshaller<T>> allocateIteratingSlot(
            int bufferCount, VFile<T> file) {
    	bufferCount = optimizeBufferCount(bufferCount, (int) file.getLengthInBytes());
        return allocateIteratingSlot(bufferCount, file.getReadableChannel(), file.getMarshaller());
    }

	private int optimizeBufferCount(int bufferCount, int lengthFileInBytes) {
		if(bufferCount * this.blockSize > lengthFileInBytes)
    		if(lengthFileInBytes > this.blockSize){
    			bufferCount = lengthFileInBytes/this.blockSize;
    			if(lengthFileInBytes%this.blockSize != 0)
    				bufferCount++;
    		}else
    			bufferCount = 1;
		return bufferCount;
	}

    public <T, M extends DataMarshaller<T>> OutputSlot<T, DataMarshaller<T>> allocateOutputSlot(
            FileChannel channel, M marshaller, boolean useLastBlock) throws IOException {
        CalculatorOfWorkingWithDisk.callAllocateOutputSlot(useLastBlock);
    	
    	if (marshaller.getDataItemSize() > this.blockSize) {
            return null;
        }
        ByteBuffer slice = allocateSlice(this.blockSize);
        if (useLastBlock) {
            readLastBlock(channel, slice);
        }
        OutputSlot<T, DataMarshaller<T>> result = new OutputSlot<T, DataMarshaller<T>>(slice, channel, marshaller);
        this.slotList.add(result);
        return result;
    }

    private void readLastBlock(FileChannel channel, ByteBuffer slice) throws IOException {
        long blockCount = channel.size() / slice.capacity();
        long position = blockCount * slice.capacity();
        channel.read(slice, position);
        channel.position(position);
    }

    /**
     * Allocates a new output slot consisting of 1 buffer for appending to {@code file}.
     * If {@code useLastFromFile} parameter
     * is {@code true}, will try to reuse available space in the last block of {@code file}, if any.
     * @throws IOException if reading from file fails.
     * @throws BufferOverflowException if there are no available buffers.
     */
    public <T, M extends DataMarshaller<T>> OutputSlot<T, DataMarshaller<T>> allocateOutputSlot(
            VFile<T> file, boolean useLastFromFile) throws IOException {
        return allocateOutputSlot(file.getFileChannel(), file.getMarshaller(), useLastFromFile);
    }

    public <T, M extends DataMarshaller<T>> OutputSlot<T, DataMarshaller<T>> allocateOutputSlot(VFile<T> file)
            throws IOException {
        return allocateOutputSlot(file.getFileChannel(), file.getMarshaller(), false);
    }

    int getBlockSize() {
        return this.blockSize;
    }

    private ByteBuffer allocateSlice(int size) {
        if (this.mainBuffer.remaining() < size) {
            tryCompact();
        }
        if (this.mainBuffer.remaining() < size) {
            throw new BufferOverflowException();
        }
        this.mainBuffer.limit(this.mainBuffer.position() + size);
        ByteBuffer slice = this.mainBuffer.slice();
        this.mainBuffer.position(this.mainBuffer.limit());
        this.mainBuffer.limit(this.mainBuffer.capacity());
        return slice;
    }

    private void tryCompact() {
        int moveOffset = 0;
        for (ListIterator<Slot> it = this.slotList.listIterator(); it.hasNext();) {
            Slot slot = it.next();
            ByteBuffer slice = slot.getSlice();
            int currentSliceStartOffset = slice.arrayOffset() - moveOffset;
            int currentSliceEndOffset = currentSliceStartOffset + slice.capacity();
            if (slot.isReleased()) {
                int remainingBytes = this.mainBuffer.capacity() - currentSliceEndOffset;
                System.arraycopy(
                        this.mainBuffer.array(), currentSliceEndOffset,
                        this.mainBuffer.array(), currentSliceStartOffset,
                        remainingBytes);
                moveOffset += slice.capacity();
                this.mainBuffer.position(currentSliceStartOffset);
                it.remove();
            }
            else if (moveOffset > 0) {
                this.mainBuffer.position(currentSliceStartOffset);
                this.mainBuffer.limit(currentSliceEndOffset);
                ByteBuffer newSlice = this.mainBuffer.slice();
                slot.setSlice(newSlice);
                this.mainBuffer.position(currentSliceEndOffset);
            }
            this.mainBuffer.limit(this.mainBuffer.capacity());
        }

    }

    public int getBufferCount() {
        return this.bufferCount;
    }
}
